package day3;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.DriverManager;
import java.sql.ResultSet;

public class Test1_RichParallelSourceFunction {



    public static void main(String[] args) throws Exception {
        //TODO 1.env-准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //TODO 2.source-加载数据
       /* DataStream<Student> studentDS = env.addSource(new MySQLSource()).setParallelism(1);

        //TODO 3.transformation-数据转换处理
        //TODO 4.sink-数据输出
        studentDS.print();*/

        //TODO 5.execute-执行
        env.execute();
    }

    //自定义source完成每隔2s查询MySQL
   /* public static class MySQLSource  extends RichParallelSourceFunction<Student> {
        Connection conn = null;
        PreparedStatement ps = null;
        //连接JDBC一次
        @Override
        public void open(Configuration parameters) throws Exception {
            //加载驱动在DriverManager源码已经有了不用写了
            conn = DriverManager.getConnection("jdbc:mysql://hadoop10:3306/test_db", "root", "123456");
            ps = conn.prepareStatement("select `id`, `name`, `age` from t_student");
        }

        private Boolean flag = true;
        @Override
        public void run(SourceContext<Student> ctx) throws Exception {
            while (flag){
                //每隔2s查询一次MySQL表
                ResultSet rs = ps.executeQuery();
                //处理结果集
                while (rs.next()){
                    int id = rs.getInt("id");
                    String name = rs.getString("name");
                    int age = rs.getInt("age");
                    ctx.collect(new Student(id,name,age));
                }
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
        //关闭JDBC一次
        @Override
        public void close() throws Exception {
            if (conn != null) conn.close();
            if (ps != null) ps.close();
        }

    }*/
}
